#include <pthread.h>
#include "utils.h"
#include "dsa/heap.h"
#include "dsa/rbtree.h"
#include "threadqueue.h"
#include "timercluster.h"

typedef struct _timernode_t
{
    struct heap_node heap_node;
    struct rb_node rb_node;
    timerid_t timer_id;
    timertype_e timer_type;
    uint32_t timer_interval;
    struct timespec expire_time;
    timercallback_t callback;
    void *userdata;
} timernode_t;

struct _timercluster_t
{
    struct heap heap;
    struct rb_root rb;
    pthread_mutex_t lock;
    pthread_cond_t cond;
    pthread_t thread;
    bool running;
};

static int compare_timer(const struct heap_node *a, const struct heap_node *b)
{
    const timernode_t *ta = (const timernode_t *)a;
    const timernode_t *tb = (const timernode_t *)b;
    return ta->expire_time.tv_sec < tb->expire_time.tv_sec ||
        (ta->expire_time.tv_sec == tb->expire_time.tv_sec &&
         ta->expire_time.tv_nsec < tb->expire_time.tv_nsec);
}

static int time_less_than(const struct timespec *a, const struct timespec *b)
{
    return a->tv_sec < b->tv_sec || (a->tv_sec == b->tv_sec && a->tv_nsec < b->tv_nsec);
}

static timernode_t *rbtree_search(struct rb_root *root, const timerid_t *timer_id)
{
    struct rb_node *n = root->rb_node;
    timernode_t *e = NULL;
    while (n)
    {
        e = rb_entry(n, timernode_t, rb_node);
        if (*timer_id < e->timer_id)
        {
            n = n->rb_left;
        }
        else if (*timer_id > e->timer_id)
        {
            n = n->rb_right;
        }
        else
        {
            return e;
        }
    }
    return NULL;
}

static int rbtree_insert(struct rb_root *root, timernode_t *entry)
{
    struct rb_node **n = &root->rb_node;
    struct rb_node *parent = NULL;
    timernode_t *e = NULL;
    while (*n)
    {
        parent = *n;
        e = rb_entry(*n, timernode_t, rb_node);
        if (entry->timer_id < e->timer_id)
        {
            n = &(*n)->rb_left;
        }
        else if (entry->timer_id > e->timer_id)
        {
            n = &(*n)->rb_right;
        }
        else
        {
            return TIMERCLUSTER_RET_ERR_OTHER;
        }
    }

    rb_link_node(&entry->rb_node, parent, n);
    rb_insert_color(&entry->rb_node, root);
    return TIMERCLUSTER_RET_SUCCESS;
}

static void *timercluster_thread_func(void *arg)
{
    timercluster_t *cluster = (timercluster_t *)arg;
    LOG_PRINT_INFO("timer cluster run");
    while (cluster->running)
    {
        pthread_mutex_lock(&cluster->lock);

        if (!cluster->heap.root)
        {
            pthread_cond_wait(&cluster->cond, &cluster->lock);
            pthread_mutex_unlock(&cluster->lock);
            continue;
        }

        timernode_t *top = (timernode_t *)cluster->heap.root;
        struct timespec now = time_mono();

        if (time_less_than(&now, &top->expire_time))
        {
            struct timespec timeout = top->expire_time;
            pthread_cond_timedwait(&cluster->cond, &cluster->lock, &timeout);
            pthread_mutex_unlock(&cluster->lock);
            continue;
        }

        heap_dequeue(&cluster->heap);
        timercallback_t cb = top->callback;
        timerid_t timer_id = top->timer_id;
        void *userdata = top->userdata;

        // LOG_PRINT_DEBUG("timer_id[%u] timeout", timer_id);
        if (top->timer_type == TIMER_TYPE_CYCLE)
        {
            top->expire_time = time_mono_after(top->timer_interval);
            heap_insert(&cluster->heap, &top->heap_node);
        }
        else
        {
            // LOG_PRINT_INFO("timer_id[%u] del success", timer_id);
            rb_erase(&top->rb_node, &cluster->rb);
            free(top);
        }

        pthread_mutex_unlock(&cluster->lock);

        if (cb)
        {
            // todo: thread pool
            cb(timer_id, userdata);
        }
    }

    LOG_PRINT_INFO("timer cluster stoped");
    return NULL;
}

timercluster_t *timercluster_init(void)
{
    timercluster_t *cluster = calloc(1, sizeof(timercluster_t));
    if (NULL == cluster)
    {
        LOG_PRINT_ERROR("calloc fail, errno[%d](%s)", errno, strerror(errno));
        return NULL;
    }
    heap_init(&cluster->heap, compare_timer);
    cluster->rb = RB_ROOT;
    pthread_mutex_init(&cluster->lock, NULL);
    pthread_cond_init(&cluster->cond, NULL);
    cluster->running = true;
    pthread_create(&cluster->thread, NULL, timercluster_thread_func, cluster);
    LOG_PRINT_INFO("timer cluster inited");

    return cluster;
}

void timercluster_destroy(timercluster_t *cluster)
{
    if (NULL == cluster)
    {
        LOG_PRINT_ERROR("invalid param!");
        return;
    }

    pthread_mutex_lock(&cluster->lock);
    cluster->running = false;
    pthread_cond_broadcast(&cluster->cond);
    pthread_mutex_unlock(&cluster->lock);
    if (cluster->thread)
    {
        pthread_join(cluster->thread, NULL);
    }

    while (cluster->heap.root)
    {
        heap_dequeue(&cluster->heap);
    }

    struct rb_node *node = NULL;
    timernode_t *entry = NULL;
    while ((node = cluster->rb.rb_node))
    {
        entry = rb_entry(node, timernode_t, rb_node);
        rb_erase(node, &cluster->rb);
        free(entry);
        entry = NULL;
    }
    pthread_mutex_destroy(&cluster->lock);
    pthread_cond_destroy(&cluster->cond);
    LOG_PRINT_WARN("timer cluster destroyed");
}

bool timercluster_timer_exist(timercluster_t *cluster, timerid_t id)
{
    if (NULL == cluster)
    {
        LOG_PRINT_ERROR("invalid param!");
        return false;
    }

    pthread_mutex_lock(&cluster->lock);
    if (!cluster->running)
    {
        LOG_PRINT_ERROR("timercluster is not running!");
        pthread_mutex_unlock(&cluster->lock);
        return false;
    }

    bool exists = (rbtree_search(&cluster->rb, &id) != NULL) ? true : false;
    pthread_mutex_unlock(&cluster->lock);

    return exists;
}

bool timercluster_timer_add(timercluster_t *cluster, timerid_t id, timertype_e type, uint32_t interval, timercallback_t cb, void *userdata)
{
    if (NULL == cluster)
    {
        LOG_PRINT_ERROR("invalid param!");
        return false;
    }

    timernode_t *tnode = NULL;
    pthread_mutex_lock(&cluster->lock);
    if (!cluster->running)
    {
        LOG_PRINT_ERROR("timercluster is not running!");
        pthread_mutex_unlock(&cluster->lock);
        return false;
    }

    tnode = rbtree_search(&cluster->rb, &id);
    if (NULL != tnode)
    {
        LOG_PRINT_ERROR("timer_id[%u] is existed", id);
        pthread_mutex_unlock(&cluster->lock);
        return false;
    }

    tnode = (timernode_t *)calloc(1, sizeof(timernode_t));
    if (NULL == tnode)
    {
        LOG_PRINT_ERROR("calloc fail, errno[%d]", errno);
        pthread_mutex_unlock(&cluster->lock);
        return false;
    }

    tnode->timer_id = id;
    tnode->timer_type = type;
    tnode->timer_interval = interval;
    tnode->expire_time = time_mono_after(interval);
    tnode->callback = cb;
    tnode->userdata = userdata;
    if (0 != rbtree_insert(&cluster->rb, tnode))
    {
        LOG_PRINT_ERROR("rbtree_insert fail!");
        pthread_mutex_unlock(&cluster->lock);
        free(tnode);
        return false;
    }
    heap_insert(&cluster->heap, &tnode->heap_node);
    pthread_cond_signal(&cluster->cond);
    pthread_mutex_unlock(&cluster->lock);

    LOG_PRINT_INFO("timer_id[%u] add success", id);

    return true;
}

bool timercluster_timer_del(timercluster_t *cluster, timerid_t id)
{
    if (NULL == cluster)
    {
        LOG_PRINT_ERROR("invalid param!");
        return false;
    }

    timernode_t *tnode = NULL;
    pthread_mutex_lock(&cluster->lock);
    if (!cluster->running)
    {
        LOG_PRINT_ERROR("timercluster is not running!");
        pthread_mutex_unlock(&cluster->lock);
        return false;
    }

    tnode = rbtree_search(&cluster->rb, &id);
    if (NULL == tnode)
    {
        LOG_PRINT_ERROR("timer_id[%u] is not existed or already deleted", id);
        pthread_mutex_unlock(&cluster->lock);
        return false;
    }

    rb_erase(&tnode->rb_node, &cluster->rb);
    heap_remove(&cluster->heap, &tnode->heap_node);
    free(tnode);
    pthread_cond_signal(&cluster->cond);
    pthread_mutex_unlock(&cluster->lock);

    LOG_PRINT_INFO("timer_id[%u] del success", id);

    return true;
}

bool timercluster_timer_reset(timercluster_t *cluster, timerid_t id, uint32_t new_interval)
{
    if (NULL == cluster)
    {
        LOG_PRINT_ERROR("invalid param!");
        return false;
    }

    timernode_t *tnode = NULL;
    pthread_mutex_lock(&cluster->lock);
    if (!cluster->running)
    {
        LOG_PRINT_ERROR("timercluster is not running!");
        pthread_mutex_unlock(&cluster->lock);
        return false;
    }

    tnode = rbtree_search(&cluster->rb, &id);
    if (NULL == tnode)
    {
        LOG_PRINT_ERROR("timer_id[%u] is not existed or already deleted", id);
        pthread_mutex_unlock(&cluster->lock);
        return false;
    }

    tnode->timer_interval = new_interval;
    tnode->expire_time = time_mono_after(new_interval);

    heap_remove(&cluster->heap, &tnode->heap_node);
    heap_insert(&cluster->heap, &tnode->heap_node);
    pthread_cond_signal(&cluster->cond);
    pthread_mutex_unlock(&cluster->lock);

    return true;
}

const char *timercluster_strerror(int err)
{
    switch (err)
    {
#define X(code, name, msg)        \
    case TIMERCLUSTER_RET_##name: \
        return msg;
        TIMERCLUSTER_FOREACH_ERR(X)
#undef X
        default:
            return "Unknown timer cluster error";
    }
}